package org.example.config.kafka;

import java.util.Properties;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.StandardEnvironment;

import com.alibaba.otter.canal.connector.kafka.consumer.CanalKafkaConsumer;

@Configuration
public class KafkaConfigruation{
    
    @Autowired
    private StandardEnvironment standardEnvironment;
    
    private static final String KAFKA_PREFIX = "kafka"; 
    
    @Bean("kafkaConfig")
    @ConfigurationProperties(prefix=KAFKA_PREFIX)
    public Properties getKafkaConfig(){
        return new Properties();
    }
    
    @Bean
    public CanalKafkaConsumer createKafka(@Qualifier("kafkaConfig") Properties properties) {
        Properties prop = new Properties();
        properties.entrySet().forEach(item->{
            prop.put(KAFKA_PREFIX+"."+item.getKey(), item.getValue());
        });
        properties.put("canal.mq.flatMessage", standardEnvironment.getProperty("canal.mq.flatMessage",Boolean.class));
        CanalKafkaConsumer consumer = new CanalKafkaConsumer();
        consumer.init(prop, properties.getProperty("topic","data-example") 
            , properties.getProperty("group.id","data-channel"));
        
        consumer.connect();
        return consumer;
    }

}
